package com.lee.trigger;

import com.lee.IP;
import com.lee.TCP;
import com.lee.UDP;
import com.lee.networkData;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;

public class TimeWindowTrigger<W extends Window> extends Trigger<networkData, W> {


    private Logger logger = LoggerFactory.getLogger(TimeWindowTrigger.class);

    HashMap<String, Integer> timeDstMap;
    HashMap<String, Integer> timeServiceMap;

    public TimeWindowTrigger(HashMap<String, Integer> timeDstMap, HashMap<String, Integer> timeServiceMap) {
        this.timeDstMap = timeDstMap;
        this.timeServiceMap = timeServiceMap;
    }

    @Override
    public TriggerResult onElement(networkData element, long timestamp, W window, TriggerContext ctx) throws Exception {
        if (!(element.getInternetLayer() instanceof IP)) {
            icmpAddOne();
            return TriggerResult.CONTINUE;
        }

        otherAddOne(element);
//        App.logger.info(map.keySet().toString());
        return TriggerResult.CONTINUE;
    }

    public void icmpAddOne() {
        if (timeServiceMap.containsKey("ICMP")) {
            timeServiceMap.put("ICMP", timeServiceMap.get("ICMP") + 1);
        } else {
            timeServiceMap.put("ICMP", 1);
        }
    }

    public void otherAddOne(networkData element) {
        IP ipProtocol = (IP) element.getInternetLayer();
        if (timeDstMap.containsKey(ipProtocol.getDstIP().toString())) {
            timeDstMap.put(ipProtocol.getDstIP().toString(), timeDstMap.get(ipProtocol.getDstIP().toString()) + 1);
        } else {
            timeDstMap.put(ipProtocol.getDstIP().toString(), 1);
        }
        if (element.getTransportLayer() == null)
            return;

        if (element.getTransportLayer() instanceof TCP) {
            TCP tcpProtocol = (TCP) element.getTransportLayer();
            String protocol = tcpProtocol.getApplicationProtocol().toString().compareTo("unknown") == 0 ? tcpProtocol.getDstPort().toString() : tcpProtocol.getApplicationProtocol().toString();
            if (timeServiceMap.containsKey(protocol)) {
                timeServiceMap.put(protocol, timeServiceMap.get(protocol) + 1);
            } else {
                timeServiceMap.put(protocol, 1);
            }
        } else {
            UDP udpProtocol = (UDP) element.getTransportLayer();
            String protocol = udpProtocol.getApplicationProtocol().toString().compareTo("unknown") == 0 ? udpProtocol.getDstPort().toString() : udpProtocol.getApplicationProtocol().toString();
            if (timeServiceMap.containsKey(protocol)) {
                timeServiceMap.put(protocol, timeServiceMap.get(protocol) + 1);
            } else {
                timeServiceMap.put(protocol, 1);
            }
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {

        return TriggerResult.FIRE_AND_PURGE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(W window, TriggerContext ctx) throws Exception {
        ctx.deleteProcessingTimeTimer(window.maxTimestamp());
    }
}
